import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by spark on 16-10-7.
 */
public class HBaseClient extends Configured implements Tool{

    // vv PageFilterExample
    private static final byte[] POSTFIX = new byte[] { 0x00 };


    @Override
    public int run(String[] args) throws Exception {

        crudTest();

//        rowFilterTest();

//        familyFilterTest();

//        pageFilterTest();

//        filterListTest();

//        customerFilterTest();


        return 0;
    }

    private void customerFilterTest() throws IOException {
        {
            Connection conn = ConnectionFactory.createConnection(getConf());
            Table table = conn.getTable(TableName.valueOf("test"));


            Filter filter = new CustomerFilter(Bytes.toBytes("value4"));
            Scan scan = new Scan();
            scan.setFilter(filter);

            int totalRows = 0;

            ResultScanner results = table.getScanner(scan);
            int localRows = 0;
            System.out.println("Scan#1----------Customer Filter-----");
            for (Result result : results) {
                System.out.println(localRows++ + " : " + result);
                totalRows++;
            }
            results.close();


            System.out.println("total rows: " + totalRows);
        }
    }

    private void filterListTest() throws IOException {


        Connection conn = ConnectionFactory.createConnection(getConf());
        Table table = conn.getTable(TableName.valueOf("test"));

        List<Filter> filters = new ArrayList<Filter>();
        filters.add(new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes("row002"))));
        filters.add(new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes("row003"))));
        filters.add(new QualifierFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryComparator(Bytes.toBytes("1"))));

        Filter filter = new FilterList(FilterList.Operator.MUST_PASS_ONE,filters);
            Scan scan = new Scan();
            scan.setFilter(filter);

        int totalRows = 0;

            ResultScanner results = table.getScanner(scan);
            int localRows = 0;
            System.out.println("Scan#1----------Multi Filter-----");
            for (Result result : results) {
                System.out.println(localRows++ + " : " + result);
                totalRows++;
            }
            results.close();


        System.out.println("total rows: " + totalRows);
    }

    private void pageFilterTest() throws IOException {

        Connection conn = ConnectionFactory.createConnection(getConf());
        Table table = conn.getTable(TableName.valueOf("test"));

        Filter filter = new PageFilter(3);

        int totalRows = 0;
        byte[] lastRow = null;
        while(true) {
            Scan scan = new Scan();
            scan.setFilter(filter);
            if (lastRow != null) {
                byte[] startRow = Bytes.add(lastRow, POSTFIX);
                System.out.println("start row: " +
                    Bytes.toStringBinary(startRow));
                scan.setStartRow(startRow);
            }


            ResultScanner results = table.getScanner(scan);
            int localRows = 0;
            System.out.println("Scan#1----------Page Filter-----");
            for (Result result : results) {
                System.out.println(localRows++ + " : " + result);
                totalRows++;
                lastRow = result.getRow();
            }
            results.close();

            if (localRows == 0)
                break;
        }

        System.out.println("total rows: " + totalRows);



    }

    private void familyFilterTest() throws IOException {

        Connection conn = ConnectionFactory.createConnection(getConf());
        Table table = conn.getTable(TableName.valueOf("test"));

        Scan scan = new Scan();
//        scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("1"));
        scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes("2"))));

        ResultScanner results = table.getScanner(scan);

        System.out.println("Scan#1----------Family Filter-----BinaryComparator");
        for (Result result : results) {
            System.out.println(result);
        }
        results.close();

        scan.setFilter(new SingleColumnValueFilter(Bytes.toBytes("data"),
                Bytes.toBytes("1"), CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("4")));
        results = table.getScanner(scan);

        System.out.println("Scan#2----------Single Column Value Filter-----RegexStringComparator");
        for (Result result : results) {
            for (Cell cell : result.listCells()) {
//                System.out.println("Cell: row = " + Bytes.toString(CellUtil.cloneRow(cell))
//                        + " /\tvalue = " + Bytes.toString(CellUtil.cloneValue(cell)));

            }
            System.out.println(result);
        }
        results.close();

        scan.setFilter(new ColumnPrefixFilter(Bytes.toBytes("2")));
        results = table.getScanner(scan);

        System.out.println("Scan#3----------Prefix Filter-----");
        for (Result result : results) {
            for (Cell cell : result.listCells()) {
                System.out.println("Cell: row = " + Bytes.toString(CellUtil.cloneRow(cell))
                        + " /\tvalue = " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }
        results.close();
    }

    private void rowFilterTest() throws IOException {
        Connection conn = ConnectionFactory.createConnection(getConf());
        Table table = conn.getTable(TableName.valueOf("test"));

        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("data"), Bytes.toBytes("1"));
        scan.setFilter(new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes("row003"))));

        ResultScanner results = table.getScanner(scan);

        System.out.println("Scan#1----------Row Filter-----BinaryComparator");
        for (Result result : results) {
            System.out.println(result);
        }
        results.close();

        scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL,
                new RegexStringComparator("row\\d{3}$")));
        results = table.getScanner(scan);

        System.out.println("Scan#2----------Row Filter-----RegexStringComparator");
        for (Result result : results) {
            System.out.println(result);
        }
        results.close();


    }

    private void crudTest() throws IOException {
        Connection conn = ConnectionFactory.createConnection(getConf());
        Admin admin = conn.getAdmin();

//        HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("joe"));
//        HColumnDescriptor hcd = new HColumnDescriptor("info");
//        hcd.setMaxVersions(3);
//        htd.addFamily(hcd);
//        admin.createTable(htd);
//        String tableName = htd.getNameAsString();
//
//        TableName[] tables = admin.listTableNames();
//
//        if (tables.length < 1 && !tables[0].getNameAsString().equals(tableName)) {
//            throw new IOException("Failed create of table");
//        }

        HTable table = (HTable) conn.getTable(TableName.valueOf("joe"));
//        table.setAutoFlushTo(false);

        List<Delete> deletes = new ArrayList<Delete>();
        Delete delete = new Delete(Bytes.toBytes("row001"));
//        delete.addColumn(Bytes.toBytes("data"), Bytes.toBytes("1"));
        deletes.add(delete);

        delete = new Delete(Bytes.toBytes("row002"));
        deletes.add(delete);

        delete = new Delete(Bytes.toBytes("row003"));
        deletes.add(delete);

        delete = new Delete(Bytes.toBytes("row004"));
        deletes.add(delete);

        table.delete(deletes);

        List<Put> puts = new ArrayList<Put>();
        byte[] row1 = Bytes.toBytes("row005");
        Put p1 = new Put(row1);
        byte[] databytes = Bytes.toBytes("info");
        p1.addColumn(databytes, Bytes.toBytes("1"), 1, Bytes.toBytes(100L));
        table.put(p1);
//        puts.add(p1);

        byte[] row2 = Bytes.toBytes("row006");
        Put p2 = new Put(row2);
        databytes = Bytes.toBytes("info");
        p2.addColumn(databytes, Bytes.toBytes("1"), 1, Bytes.toBytes(23L));
        table.put(p2);
//        puts.add(p2);

        p2 = new Put(Bytes.toBytes("row007"));
        p2.addColumn(databytes, Bytes.toBytes("2"), 1, Bytes.toBytes(999L));
        p2.addColumn(databytes, Bytes.toBytes("3"), 1, Bytes.toBytes(99L));
        table.put(p2);
//        puts.add(p2);


        byte[] row3 = Bytes.toBytes("row008");
        Put p3 = new Put(row3);
        databytes = Bytes.toBytes("info");
        p3.addColumn(databytes, Bytes.toBytes("1"), 1, Bytes.toBytes(121L));
        p3.addColumn(databytes, Bytes.toBytes("2"), 1, Bytes.toBytes(21L));
        table.put(p3);
//        puts.add(p3);

//        table.put(puts);

//        table.checkAndPut(row3, Bytes.toBytes("data"), Bytes.toBytes("1"), null, p3);

//        table.flushCommits();

        Get g = new Get(row1);
//        g.setMaxResultsPerColumnFamily(10);


        Result result = table.get(g);
        System.out.println("Get: " + result);

//        Delete delete = new Delete(Bytes.toBytes("row2"));
//        delete.addColumn(Bytes.toBytes("data"), Bytes.toBytes("1"));
//        table.delete(delete);

        Scan scan = new Scan();
        scan.setMaxVersions();
        ResultScanner scanner = table.getScanner(scan);
        for (Result scannerResult:
                scanner) {
            System.out.println("Scan: " + scannerResult);
        }
        scanner.close();

//        admin.disableTable(TableName.valueOf(tableName));
//        admin.deleteTable(TableName.valueOf(tableName));

    }


    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new HBaseClient(), args);
        System.exit(exitCode);
    }
}
